package com.wichell.framework.rocketmq.example.ordermessage;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class OrderConsumer {
   public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsumer");
       consumer.setNamesrvAddr("10.138.61.57:9876;10.138.61.59:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
       // consumer.setConsumeThreadMin(1);
        consumer.subscribe("TopicOrder", "*");
       // consumer.setSuspendCurrentQueueTimeMillis(10000);

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                for(MessageExt messageExt : msgs){try{
                 String msgBody = new String(messageExt.getBody(),"utf-8");
                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + messageExt + "   messageBody:" + msgBody);
               
                }catch(Exception e){
                	e.printStackTrace();
                	}
                }
               
                return ConsumeOrderlyStatus.SUCCESS;

            }
        });

        consumer.start();
      
        System.out.println("Order Consumer Started");
       /* try{
        Thread.sleep(1000000);
        }catch(Exception e){
        	e.printStackTrace();
        }*/
    }
}